"""
Case Type   : Logical_Replication
Case Name   : pg_logical_slot_peek_binary_changes函数解码验证，lsn不推进
Create At   : 2024/1/15
Owner       : xiaqi14
Description :
    1.修改参数
    2.配置初始用户白名单
    3.创建表
    4.创建逻辑复制槽
    5.向表中插入数据，在主机上执行此函数，查询lsn
    6.向表中更新数据，在备机上执行此函数，查询lsn
    7.删除逻辑复制槽，删除表
    8.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.配置成功
    3.创建成功
    4.创建成功
    5.插入数据成功，解码成功，lsn不推进
    6.更新数据成功，解码2条sql成功，lsn不推进
    7.删除成功
    8.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == primary_sh.get_node_num(), '单机环境不执行')
class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.table_name = 't_logical_replication_0071'
        self.slot_name = 'slot_logical_replication_0071'
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'on' != self.slot_log_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
            f'::1/128 trust'
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:创建表;expect:创建成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
            create table {self.table_name}(
                col_integer integer,
                col_date date,
                col_varchar varchar(20));'''
        create_msg = primary_sh.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      create_msg, '执行失败:' + text)

        text = "-----step4:创建逻辑复制槽;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select pg_create_logical_replication_slot(" \
            f"'{self.slot_name}', 'mppdb_decoding');" \
            f"select restart_lsn from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name}%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name, sql_msg.splitlines()[2],
                      '执行失败:' + text)
        restart_lsn = sql_msg.splitlines()[-2].strip()
        self.logger.info(restart_lsn)

        text = "-----step5:向表中插入数据，在主机上执行此函数，查询lsn，" \
               "expect:插入数据成功，解码成功，lsn不推进-----"
        self.logger.info(text)
        sql_cmd = f"insert into {self.table_name} " \
            f"values(1,date '2022-05-22','测试');" \
            f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name}', '', 3,'include-xids', '1');" \
            f"select restart_lsn from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name}%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG and
                      "pg_logical_slot_peek_binary_changes" and
                      "(3 rows)" and restart_lsn,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:向表中更新数据，在备机上执行此函数，" \
               "expect:更新数据成功-----"
        self.logger.info(text)
        sql_cmd = f"update {self.table_name} set " \
            f"col_integer=2,col_varchar='开发';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)

    def tearDown(self):
        text_1 = "-----step7:删除逻辑复制槽，删除表;expect:删除成功-----"
        self.logger.info(text_1)
        sql_cmd = f"select pg_drop_replication_slot('{self.slot_name}');" \
            f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name}%';" \
            f"drop table {self.table_name};"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_2 = "-----step8:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_2)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        self.assertIn("(0 rows)", sql_msg.splitlines()[-3], '执行失败:' + text_1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      sql_msg.splitlines()[-1], '执行失败:' + text_1)
        self.assertEquals(show_slot_log, self.slot_log_default,
                          '执行失败' + text_2)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
